package com.gitee.xhs.broadcastState;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.sampleSource.TransactionUnBoundOneSecondSource;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static com.gitee.xhs.broadcastState.BroadcastStateConstant.*;

/**
 * @author BigHead
 * @date 2022/7/6 17:19
 * @description BroadcastStateMain
 */
public class BroadcastStateMain {

    public static void main(String[] args) throws Exception {
        EnvironmentConfig environmentConfiguration = new EnvironmentConfig(null);
        environmentConfiguration.checkpoint(
                null,
                null,
                BROADCAST_STATE_PATH,
                null
        );

        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.simpleEnv(environmentConfiguration);

        // 主流
        DataStream<Transaction> transactionMainStream = streamExecutionEnvironment
                .addSource(new TransactionUnBoundOneSecondSource());

        // 广播流
        BroadcastStream<BroadcastStateDao> broadcastStateDaoBroadcastStream = streamExecutionEnvironment
                .addSource(new BroadcastStateSource())
                .broadcast(BROADCAST_STATE_MAP_STATE_DESCRIPTOR);

        // 合并广播流
        transactionMainStream
                .connect(broadcastStateDaoBroadcastStream)
                .process(new BroadcastStateBroadcastProcessFunction()).uid(BROADCAST_STATE_UID)
                .print();

        streamExecutionEnvironment.execute("BroadcastStateMain");
    }

}
